package com.leo.hbase.mapreduce.service;

import com.leo.hbase.mapreduce.exception.DbException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.IOException;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization CC
 * @website https://www.jlpyyf.com
 * @date 2019/11/13 9:59 下午
 * @since 1.0
 */
public class HbaseClient extends AbstractDbService {
    private static final AtomicInteger THREAD_COUNT = new AtomicInteger(0);
    private Configuration config = HBaseConfiguration.create();
    private boolean debug = false;

    private String tableName = "";

    private static Connection connection = null;
    private Table currentTable = null;
    private BufferedMutator bufferedMutator = null;

    private boolean usePageFilter = true;
    /**
     * 如果为true，则在客户端上缓冲突变。 这是默认行为
     * HBaseClient。 用于测量插入/更新/删除延迟，客户端
     * 应该禁用缓冲。
     */
    private boolean clientSideBuffering = false;
    private long writeBufferSize = 1024 * 1024 * 12;

    @Override
    public void init() throws DbException {
        if ("true".equals(getProperties().getProperty("clientBuffering", "false"))) {
            this.clientSideBuffering = true;
        }
        if (getProperties().containsKey("writeBufferSize")) {
            this.writeBufferSize = Long.parseLong(getProperties().getProperty("writeBufferSize"));
        }
        if ("kerberos".equalsIgnoreCase(config.get("hbase.security.authentication"))) {
            config.set("hbase.security.authentication", "Kerberos");
            UserGroupInformation.setConfiguration(config);
        }
        if ((getProperties().getProperty("principal") != null)
                && (getProperties().getProperty("keytab") != null)) {
            try {
                UserGroupInformation.loginUserFromKeytab(getProperties().getProperty("principal"),
                        getProperties().getProperty("keytab"));
            } catch (IOException e) {
                System.err.println("Keytab file 不可用或找不到！");
                throw new DbException(e);
            }
        }
        try {
            THREAD_COUNT.getAndIncrement();
            synchronized (THREAD_COUNT) {
                if (connection == null) {
                    //初始化连接
                    connection = ConnectionFactory.createConnection(config);
                }
            }
        } catch (IOException e) {
            throw new DbException(e);
        }
        if ((getProperties().getProperty("debug") != null)
                && (getProperties().getProperty("debug").compareTo("true") == 0)) {
            this.debug = true;
        }
    }

    @Override
    public void cleanup() throws DbException {
        try {
            int threadCount = THREAD_COUNT.decrementAndGet();
            if (threadCount <= 0) {
                synchronized (THREAD_COUNT) {
                    if (connection != null) {
                        connection.close();
                        connection = null;
                    }
                }
            }
        } catch (IOException e) {
            throw new DbException(e);
        }
    }

    public void getHTable(String table) throws IOException {
        final TableName tName = TableName.valueOf(table);
        this.currentTable = connection.getTable(tName);
        if (clientSideBuffering) {
            final BufferedMutatorParams p = new BufferedMutatorParams(tName);
            p.writeBufferSize(writeBufferSize);
            this.bufferedMutator = connection.getBufferedMutator(p);
        }
    }


    @Override
    public Map<String, ByteIterator> read(String table, String columnFamily, String key, Set<String> fields) {
        Map<String, ByteIterator> result = new HashMap<>(8);
        if (!tableName.equals(table)) {
            currentTable = null;
            try {
                getHTable(table);
                this.tableName = table;
            } catch (IOException e) {
                System.out.println("Hbase表异常：" + e);
                return result;
            }
        }
        Result r;
        try {
            if (debug) {
                System.out.println("正在读取Hbase的列簇: " + columnFamily);
                System.out.println("正在读取的key: " + key);
            }
            Get g = new Get(Bytes.toBytes(key));
            byte[] columnFamilyBytes = Bytes.toBytes(columnFamily);
            if (fields == null) {
                g.addFamily(columnFamilyBytes);
            } else {
                for (String field : fields) {
                    g.addColumn(columnFamilyBytes, Bytes.toBytes(field));
                }
            }
            r = currentTable.get(g);
        } catch (IOException e) {
            if (debug) {
                System.err.println("Read Row Key: " + e);
            }
            return result;
        } catch (ConcurrentModificationException e) {
            return result;
        }
        if (r.isEmpty()) {
            return result;
        }
        while (r.advance()) {
            final Cell c = r.current();
            result.put(Bytes.toString(CellUtil.cloneQualifier(c)),
                    new ByteArrayByteIterator(CellUtil.cloneValue(c)));
        }
        return result;
    }

}
